package com.zhou.cloud.producer.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;

/**
 * @descrition:
 * @author: zhouyongjun
 * @date: 2022/7/28
 **/
@Service
@Slf4j
public class KafkaProducerService {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void sendMessage( String normalMessage) {
        String sendTopic = "topic1";
        kafkaTemplate.send(sendTopic, normalMessage);
        log.warn("生产消息：topic={},message={}",sendTopic,normalMessage);
    }

    /**
    * @Description: 带回调的生产者
    * @Params:
    * @Returns:
    * @Author: zhouyongjun
    * @Date Created in 2022/7/28 17:13
     */
    public void sendMessageCallBack( String normalMessage) {
        String sendTopic = "topic1";
        kafkaTemplate.send(sendTopic, normalMessage).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });
        log.warn("生产消息：topic={},message={}",sendTopic,normalMessage);
    }

    /**
    * @Description: 带成功和失败的回调
    * @Params: 
    * @Returns: 
    * @Author: zhouyongjun
    * @Date Created in 2022/7/28 17:26
     */
    public void sendMessageCallBack2(String callbackMessage) {
        kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败："+ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
    }

    /**
     * @Description: kafka事务提交
     * 如果在发送消息时需要创建事务，可以使用 KafkaTemplate 的 executeInTransaction 方法来声明事务，
     * 使用事务注意配置文件修改 application.yml：
     * 1.加事务前缀，自动给producer开启事务
     * # 事务前缀
     * spring.kafka.producer.transaction-id-prefix=tx_
     * 2.使用事务会出现 幂等情况 https://blog.csdn.net/qq_34475529/article/details/122184604
     *      如果用户还显式地指定了 acks 参数，那么还需要保证这个参数的值为 -1（all），如果不为 -1（这个参数的值默认为1），那么也会报出 ConfigException：
     * @Params:
     * @Returns:
     * @Author: zhouyongjun
     * @Date Created in 2022/7/28 17:26
     */
    public void sendMessageInTransaction(String message){
        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("topic1","test executeInTransaction");
            throw new RuntimeException("fail");
        });

        // 不声明事务：后面报错但前面消息已经发送成功了
        kafkaTemplate.send("topic1",message);
        throw new RuntimeException("fail");
    }
}
